package org.zjt.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.ReduceOperator;
import org.apache.flink.util.Collector;
import org.zjt.flink.wordcount.util.WordCountData;
import scala.Tuple2;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-09-29 下午12:04
 * @see
 */
public class WordCountPojo {
    public static class Word {

        // fields
        private String word;
        private int frequency;

        // constructors
        public Word() {}

        public Word(String word, int i) {
            this.word = word;
            this.frequency = i;
        }

        // getters setters
        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public int getFrequency() {
            return frequency;
        }

        public void setFrequency(int frequency) {
            this.frequency = frequency;
        }

        @Override
        public String toString() {
            return "Word=" + word + " freq=" + frequency;
        }
    }


    public static void main(String[] args) throws Exception {
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

        ReduceOperator<Word> word = executionEnvironment.fromElements(WordCountData.WORDS).flatMap(new TextLineSplitter()).groupBy("word").reduce((t1, t2) -> {
            t1.setFrequency(t1.getFrequency() + t2.getFrequency());
            return t1;
        });
        word.print();
    }

    private static class TextLineSplitter implements FlatMapFunction<String, Word> {
        @Override
        public void flatMap(String s, Collector<Word> collector) throws Exception {
            String[] tokens = s.toLowerCase().split("\\W+");
            for (String token: tokens) {
                if (token.length() > 0) {
                    collector.collect(new Word(token,1));
                }
            }
        }
    }
}
